package com.example.asyncio;

import com.alibaba.fastjson.JSONObject;
import com.example.entity.Customer;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

/**
 * Author wangJinLong
 * Date 2025/8/18 10:54
 **/
@Slf4j
public class AsyncDatabaseRequest extends RichAsyncFunction<Customer, Customer> {

    /** 能够利用回调函数并发发送请求的数据库客户端 */
    private transient DatabaseClient client;

    @Override
    public void open(OpenContext openContext) throws Exception {
        log.info("AsyncDatabaseRequest open");
        client = new DatabaseClient();
    }

    @Override
    public void close() throws Exception {
        log.info("AsyncDatabaseRequest close");
        client.close();
    }

    @Override
    public void timeout(Customer input, ResultFuture<Customer> resultFuture) throws Exception {
        resultFuture.complete(Collections.singleton(input));
    }

    @Override
    public void asyncInvoke(Customer customer, ResultFuture<Customer> resultFuture) throws Exception {
        log.info("AsyncDatabaseRequest asyncInvoke " + customer.getFeaturesId());
        FutureTask<Customer> result = client.queryAsyncDb(customer.getFeaturesId());

        // 设置客户端完成请求后要执行的回调函数
        // 回调函数只是简单地把结果发给 future
        CompletableFuture.supplyAsync(() -> {
            try {
                return result.get();
            } catch (InterruptedException | ExecutionException e) {
                // 显示地处理异常。
                e.printStackTrace();
                return new Customer();
            }
        }).thenAccept( (Customer customer1) -> {
            log.info("AsyncDatabaseRequest thenAccept " + JSONObject.toJSONString(customer1));

            Customer customer2 = new Customer();
            customer2.setFeaturesId(customer1.getFeaturesId());
            customer2.setSiteSn(customer1.getSiteSn());
            customer2.setCaptureTime(customer1.getCaptureTime());

            resultFuture.complete(Collections.singleton(customer2));
        });
    }
}
